package com.hmall.item.listener;

import cn.hutool.json.JSONUtil;
import com.hmall.common.constant.EsConstant;
import com.hmall.common.constant.MQConstant;
import com.hmall.item.domain.doc.ItemDoc;
import com.hmall.item.domain.dto.ItemSyncMsgDto;
import com.hmall.item.domain.po.Item;
import com.hmall.item.service.IItemService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Objects;

@Slf4j
//@Component
public class ItemSyncListener {

    @Autowired
    IItemService itemService;
    @Autowired
    RestHighLevelClient client;

    @RabbitListener(bindings = @QueueBinding(
            value=@Queue(name = MQConstant.ITEM_SYNC_QUEUE),
            exchange = @Exchange(name = MQConstant.ITEM_SYNC_TOPIC, type = ExchangeTypes.DIRECT),
            key = MQConstant.ITEM_SYNC_KEY
    ))
    public void onItemSyncMsg(String msg){
        log.info("ItemSyncListener msg：{}", msg);
        ItemSyncMsgDto dto = JSONUtil.toBean(msg, ItemSyncMsgDto.class);
        Integer type = dto.getType();
        if(Objects.equals(type, ItemSyncMsgDto.ADD) || Objects.equals(type, ItemSyncMsgDto.UPDATE)){
            // 新增或者是修改
            BulkRequest bulkRequest= new BulkRequest(EsConstant.INDEX);
            List<Long> ids = dto.getIds();
            List<Item> items = itemService.listByIds(ids);
            if(CollectionUtils.isEmpty(items)){
                log.error("根据ids查询item失败，ids：{}", ids);
                return;
            }
            for (Item item : items) {
                ItemDoc itemDoc = ItemDoc.of(item);
                IndexRequest indexRequest = new IndexRequest(EsConstant.INDEX).id(""+item.getId());
                indexRequest.source(JSONUtil.toJsonStr(itemDoc), XContentType.JSON);
                bulkRequest.add(indexRequest);
            }
            try{
                BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
                if(response.hasFailures()){
                    log.error(response.buildFailureMessage());
                }else{
                    log.info("同步成功，ids:{}", ids);
                }
            }catch(Exception e){
                log.error(e.getMessage(), e);
            }
        }else{
            // 删除
            BulkRequest bulkRequest= new BulkRequest(EsConstant.INDEX);
            List<Long> ids = dto.getIds();
            for (Long id : ids) {
                DeleteRequest deleteRequest = new DeleteRequest(EsConstant.INDEX, ""+id);
                bulkRequest.add(deleteRequest);
            }
            try{
                BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
                if(response.hasFailures()){
                    log.error(response.buildFailureMessage());
                }else{
                    log.info("同步成功，ids:{}", ids);
                }
            }catch(Exception e){
                log.error(e.getMessage(), e);
            }
        }
    }

}